package com.bujunjie.study.realtime.dws.app;

import cn.hutool.core.date.DatePattern;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import com.bujunjie.study.realtime.common.base.BaseApp;
import com.bujunjie.study.realtime.common.constant.FlinkConstant;
import com.bujunjie.study.realtime.common.function.Object2JsonStringFunction;
import com.bujunjie.study.realtime.common.model.sink.PageInfoModel;
import com.bujunjie.study.realtime.common.util.FlinkSinkUtil;
import com.bujunjie.study.realtime.dws.app.model.sink.TrafficHomeDetailPageViewModel;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AllWindowedStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Arrays;
import java.util.Date;

/**
 * <p>页面详情</p>
 *
 * @author bu.junjie
 * @version 1.0.0
 * @createTime 2025/9/22 9:59
 */
public class DwsTrafficHomeDetailPageViewWindow extends BaseApp {

    public static void main(String[] args) throws Exception {
        new DwsTrafficHomeDetailPageViewWindow().start(10023, 4, "dws_traffic_home_detail_page_view_window", FlinkConstant.TOPIC_DWD_TRAFFIC_PAGE);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> kafkaStrDS) {
        SingleOutputStreamOperator<PageInfoModel> filterDS = kafkaStrDS.process(new ProcessFunction<String, PageInfoModel>() {
            @Override
            public void processElement(String value, ProcessFunction<String, PageInfoModel>.Context ctx, Collector<PageInfoModel> out) throws Exception {
                PageInfoModel pageInfoModel = JSON.parseObject(value, new TypeReference<PageInfoModel>() {
                });
                String pageId = pageInfoModel.getPage().getPageId();
                if (Arrays.asList("home", "good_detail").contains(pageId)) {
                    out.collect(pageInfoModel);
                }
            }
        });
        //  指定 水印
        SingleOutputStreamOperator<PageInfoModel> watermarkDS = filterDS.assignTimestampsAndWatermarks(WatermarkStrategy
                .<PageInfoModel>forBoundedOutOfOrderness(Duration.ofSeconds(2 * 60))
                .withTimestampAssigner(new SerializableTimestampAssigner<PageInfoModel>() {
                    @Override
                    public long extractTimestamp(PageInfoModel pageInfoModel, long recordTimestamp) {
                        return pageInfoModel.getTs();
                    }
                }));

        KeyedStream<PageInfoModel, String> midKeyDS = watermarkDS.keyBy(x -> x.getCommon().getMid());
        // 判断当前是不是独立访客

        SingleOutputStreamOperator<TrafficHomeDetailPageViewModel> dealDS = midKeyDS.process(new KeyedProcessFunction<String, PageInfoModel, TrafficHomeDetailPageViewModel>() {

            /**
             * 上一次访问状态
             */
            private ValueState<Long> homeLastVisitDateState;
            /**
             * 详情页上次的访问状态
             */
            private ValueState<Long> detailLastVisitDateState;


            @Override
            public void open(Configuration parameters) throws Exception {
                ValueStateDescriptor<Long> homeLastVisitDescriptor = new ValueStateDescriptor<>("home-last-visit-date", Long.class);
                homeLastVisitDateState = getRuntimeContext().getState(homeLastVisitDescriptor);
                ValueStateDescriptor<Long> detailLastVisitDescriptor = new ValueStateDescriptor<>("detail-last-visit-date", Long.class);
                detailLastVisitDateState = getRuntimeContext().getState(detailLastVisitDescriptor);
            }

            @Override
            public void processElement(PageInfoModel pageInfoModel, KeyedProcessFunction<String, PageInfoModel, TrafficHomeDetailPageViewModel>.Context ctx, Collector<TrafficHomeDetailPageViewModel> out) throws Exception {
                Long ts = pageInfoModel.getTs();
                Long homeUvCt = 0L;
                Long detailUvCt = 0L;
                String pageId = pageInfoModel.getPage().getPageId();
                if ("home".equals(pageId)) {
                    Long homeLastVisitDate = homeLastVisitDateState.value();
                    if (ObjectUtil.isNull(homeLastVisitDate) || !homeLastVisitDate.equals(pageInfoModel.getTs())) {
                        homeUvCt = 1L;
                        homeLastVisitDateState.update(ts);
                    }
                } else {
                    Long detailLastVisitDate = detailLastVisitDateState.value();
                    if (ObjectUtil.isNull(detailLastVisitDate) || !detailLastVisitDate.equals(pageInfoModel.getTs())) {
                        detailUvCt = 1L;
                        detailLastVisitDateState.update(ts);
                    }
                }
                Long zero = Long.parseLong(FlinkConstant.ZERO);

                if (!zero.equals(homeUvCt) || !zero.equals(detailUvCt)) {
                    TrafficHomeDetailPageViewModel build = TrafficHomeDetailPageViewModel.builder()
                            .stt(StrUtil.EMPTY)
                            .edt(StrUtil.EMPTY)
                            .curDate(StrUtil.EMPTY)
                            .homeUvCt(homeUvCt)
                            .goodDetailUvCt(detailUvCt)
                            .ts(ts)
                            .build();
                    out.collect(build);
                }
            }
        });

        // 开窗
        AllWindowedStream<TrafficHomeDetailPageViewModel, TimeWindow> windowDS = dealDS.windowAll(TumblingEventTimeWindows.of(Time.seconds(10)));

        SingleOutputStreamOperator<TrafficHomeDetailPageViewModel> reduceDS = windowDS.reduce(new ReduceFunction<TrafficHomeDetailPageViewModel>() {
            @Override
            public TrafficHomeDetailPageViewModel reduce(TrafficHomeDetailPageViewModel value1, TrafficHomeDetailPageViewModel value2) throws Exception {
                value1.setHomeUvCt(value1.getHomeUvCt() + value2.getHomeUvCt());
                value1.setGoodDetailUvCt(value1.getGoodDetailUvCt() + value2.getGoodDetailUvCt());
                return value1;
            }
        }, new AllWindowFunction<TrafficHomeDetailPageViewModel, TrafficHomeDetailPageViewModel, TimeWindow>() {
            @Override
            public void apply(TimeWindow window, Iterable<TrafficHomeDetailPageViewModel> values, Collector<TrafficHomeDetailPageViewModel> out) throws Exception {
                TrafficHomeDetailPageViewModel viewBean = values.iterator().next();
                String stt = DateUtil.format(new Date(window.getStart()), DatePattern.NORM_DATETIME_PATTERN);
                String edt = DateUtil.format(new Date(window.getEnd()), DatePattern.NORM_DATETIME_PATTERN);
                String curDate = DateUtil.format(new Date(window.getEnd()), DatePattern.NORM_DATE_PATTERN);
                viewBean.setStt(stt);
                viewBean.setEdt(edt);
                viewBean.setCurDate(curDate);
                out.collect(viewBean);
            }
        });

        // 8.将聚合的结果写到Doris
        reduceDS.print();
        reduceDS
                .map(new Object2JsonStringFunction<TrafficHomeDetailPageViewModel>())
                .sinkTo(FlinkSinkUtil.getDorisSink("dws_traffic_home_detail_page_view_window"));


    }
}
